package com.audaque.springboot.foshanupload.web.flinkdemo.main;

import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;
import lombok.ToString;
import org.apache.flink.api.common.RuntimeExecutionMode;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.triggers.ContinuousEventTimeTrigger;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

import java.time.Duration;

public class FlinkWindowT {
    public static void main(String[] args) throws Exception {
        // 1.准备环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        env.setRuntimeMode(RuntimeExecutionMode.STREAMING);
        //2.Source: nc -l -p 9999
        DataStreamSource<String> socketDS = env.socketTextStream("172.16.101.119", 9999);
        // 数据格式 1 1 999
        SingleOutputStreamOperator<MyEntity> tp = socketDS.map(new MapFunction<String, MyEntity>() {
            @Override
            public MyEntity map(String value) {
                String[] s = value.split(" ");
                MyEntity myEntity = new MyEntity();
                if (s.length < 2) {
                    return myEntity;
                }
                myEntity = new MyEntity(s[0], Integer.parseInt(s[1]), Long.parseLong(s[2]));
                return myEntity;
            }
        });
        //水印,左闭右开,延迟窗口关闭时间
        WatermarkStrategy<MyEntity> watermarkStrategy = WatermarkStrategy
                .<MyEntity>forBoundedOutOfOrderness(Duration.ofMillis(3))
                .withTimestampAssigner((entity, xxx) -> entity.timestamp);
        //数据设置水印
        SingleOutputStreamOperator<MyEntity> singleOutputStreamOperator = tp.assignTimestampsAndWatermarks(watermarkStrategy);
        //定义侧输出流
        OutputTag<MyEntity> myEntityOutputTag = new OutputTag<MyEntity>("lose") {
        };
        SingleOutputStreamOperator<MyClass> process = singleOutputStreamOperator.keyBy(myEntity -> myEntity.id)
                //设置窗口时间间隔
                .window(TumblingEventTimeWindows.of(Time.milliseconds(20)))
                //设置触发窗口计算的间隔
                .trigger(ContinuousEventTimeTrigger.of(Time.milliseconds(2)))
                //延迟销毁窗口;默认触发下一个窗口后上一个窗口会销毁,后面再有上一个时间的窗口的数据会被丢掉;在设置延迟销毁后,在延迟销毁时间内每一个迟到的元素进入窗口，都会触发一次窗口计算
                .allowedLateness(Time.milliseconds(3))
                //设置侧输出流
                .sideOutputLateData(myEntityOutputTag)
                .process(new ProcessFunction());
        //测输出流处理
        process.getSideOutput(myEntityOutputTag).print();

        process.print();
//        process.addSink(new MyMongoSink());
        env.execute();

    }

    private static class ProcessFunction extends ProcessWindowFunction<MyEntity, MyClass, String, TimeWindow> {

        public ProcessFunction() {
        }

        @Override
        public void process(String id, ProcessWindowFunction<MyEntity, MyClass, String, TimeWindow>.Context context, Iterable<MyEntity> elements, Collector<MyClass> out) throws Exception {
            int cnt = 0;
            for (MyEntity element : elements) {
                cnt += element.count;
                System.out.println(element);
            }
            out.collect(new MyClass(id, cnt));
            System.out.println("----------------------------------------");
        }
    }


    @ToString
    @Data
    @AllArgsConstructor
    @NoArgsConstructor
    public static class MyEntity {
        private String id;
        private Integer count;
        // 单位毫秒
        private Long timestamp;
    }
    @ToString
    @Data
    @AllArgsConstructor
    @NoArgsConstructor
    public static class MyClass {
        private String id;
        private Integer count;
    }
}
